import datetime
import json
import pymysql
class MySQLHelper:
    def __init__(self, host, port, user, password, db, charset='utf8'):
        self.host = host
        self.port = port
        self.user = user
        self.password = password
        self.db = db
        self.charset = charset
        self.conn = None
        self.cursor = None
    def connect(self):
        self.conn = pymysql.Connect(host=self.host, port=self.port,
                                    user=self.user, password=self.password,
                                    db=self.db, charset=self.charset)
        self.cursor = self.conn.cursor()

    def execute_select(self, sql):
        self.connect()
        self.cursor.execute(sql)
        ret = self.cursor.fetchall()
        data = []
        for k in ret:
            target = {}
            for j in range(len(self.cursor.description)):
                target[self.cursor.description[j][0]] = k[j]
            data.append(target)
        self.close()
        return data

    def execute_edit(self, sql):
        self.connect()
        ret = self.cursor.execute(sql)
        self.conn.commit()
        self.close()
        return ret

    def execute_charts(self, sql):
        self.connect()
        self.cursor.execute(sql)
        ret = self.cursor.fetchall()
        target = {}
        for j in range(len(self.cursor.description)):
            data = []
            for k in ret:
                data.append(k[j])
            target[self.cursor.description[j][0]] =data
        self.close()
        return target

    def execute_datatable(self, sql):
        self.connect()
        self.cursor.execute(sql)
        ret = self.cursor.fetchall()
        datas = []
        for k in ret:
            target = []
            for j in range(len(self.cursor.description)):
                target.append( k[j])
            datas.append(target)
        data = {"data":datas}
        self.close()
        return data

    def execute_cloud(self, sql):
        self.connect()
        self.cursor.execute(sql)
        ret = self.cursor.fetchall()
        data={}
        for k in ret:
            data[k[0]]=k[1]
        self.close()
        return data

    def close(self):
        if self.cursor is not None:
            self.cursor.close()
        if self.conn is not None:
            self.conn.close()


import pymysql
import redis
import pandas
db = MySQLHelper(host='mmservice-05.mysql.hotgrid.cn', port=3306,
                 user='electricity_api_service', password='GJlfh7&#jg',
                 db='electricity_data', charset='utf8')
class DailyAlarmNewRule2():

    def delete_data_elec_warn_records(self,times):
        import time

        print(times,"当前时间")
        sql = f"""DELETE ewr 
        FROM elec_warn_records ewr 
        JOIN (
        SELECT id 
        FROM elec_warn_records
        WHERE start_time = '{times}' ) t ON ewr.id = t.id"""
        res=db.execute_edit(sql)
        time.sleep(3)
        if res:
            print(times + "删除成功")
        return "ok"

    def delete_data_elec_daily_alarm_rule2(self,time):
        sql = f"""DELETE ewr 
        FROM elec_daily_alarm_rule2 ewr 
        JOIN (
        SELECT id 
        FROM elec_daily_alarm_rule2
        WHERE data_time = '{time}' ) t ON ewr.id = t.id"""
        res=db.execute_edit(sql)
        print(res)
        if res:
            print(time+"删除成功")
        return "ok"


    def run(self, data=""):
        self.cur, self.conn = self._db_util()
        self.county_id, delta,self.data_time  = str(data).split(',')
        # global CUR_TIME
        # cur_time = datetime.datetime(2023,9,18,17,37,0)
        # cur_time = datetime.datetime(2023,10,16,15,0,0)
        cur_time = datetime.datetime.now()
        # cur_time = datetime.datetime(2023,9,8,14,44,0)
        # cur_time = CUR_TIME

        cur_min = 15 * (cur_time.minute // 15)
        if self.data_time == '0':
            self.data_time = (datetime.datetime(cur_time.year,
                                                cur_time.month,
                                                cur_time.day,
                                                cur_time.hour,
                                                cur_min, 00)
                              - datetime.timedelta(minutes=(int(delta)+5) * 15)).strftime('%Y-%m-%d %H:%M:00')
        # todo 针对回算的情形，还需考虑latest_alarm与>data_time的报警的交集部分，避免重复
        latest_alarm = self.get_latest_alarm()
        unreleased_warn = self.get_unreleased_warn()
        inter_part = self.get_inter_dev_info(latest_alarm, unreleased_warn)
        print(self.data_time,"时间有问题")
            # 如果出问题 删除这段时间的数据
        sql=f"""DELETE ewr 
FROM elec_warn_records ewr 
JOIN (
SELECT id 
FROM elec_warn_records
WHERE start_time = '{self.data_time}' ) t ON ewr.id = t.id"""
        db.execute_edit(sql)
        print(latest_alarm,inter_part,"插入报警 之前所需要获得的数据")
        self.insert_new_warn(latest_alarm, inter_part)
        self.update_unreleased_warn(unreleased_warn, inter_part,latest_alarm)


    def get_latest_alarm(self):
        sql = f'''select data_time, ent_id,dev_id,dev_name, yield_dev_id_list,
        ent_name, industry_type_id,industry_type, county_id, 
        county_name, town_id, town_name from elec_daily_alarm_rule2 
                    where data_time='{self.data_time}' and is_deleted=0'''
        res_pd = self._db_res_to_df(sql)

        return res_pd

    # todo 针对回算的情形，<=end_time这个条件不够严谨
    def get_unreleased_warn(self):
        sql = f'''select code, warn_level, warn_type, 
        start_time, end_time, duration,
        release_status,dev_id, dev_name, dev_type, 
        dev_type_name, other_data, ent_id, ent_name, industry_type_id, 
        industry_type, county_id, county_name, town_id, town_name, rule_number 
        from elec_warn_records
        where is_deleted=0 and rule_number=2 and warn_type=4
        and release_status=0 and end_time<="{self.data_time}"'''
        res_pd = self._db_res_to_df(sql)

        return res_pd

    def get_inter_dev_info(self, latest_alarm, unreleased_warn):
        if latest_alarm.empty or unreleased_warn.empty:
            return pandas.DataFrame(columns=['ent_id', 'dev_id','county_id'])

        # 先提取出交集部分
        a = latest_alarm[['ent_id', 'dev_id','county_id']].drop_duplicates()
        b = unreleased_warn[['ent_id', 'dev_id','county_id']].drop_duplicates()
        c = pandas.merge(a, b, on=['ent_id', 'dev_id','county_id'])

        return c

    def get_diff_part(self, df1, df2, df3):
        a = df1[['ent_id', 'county_id', 'dev_id']]
        if df2.empty:
            b = pandas.DataFrame()
        else:
            b = df2[['ent_id', 'county_id', 'dev_id']]
        a = a.append(b)
        a = a.append(b)
        c = a.drop_duplicates(subset=a.columns, keep=False)

        res = pandas.merge(df3, c, on=['ent_id', 'county_id', 'dev_id'], how='right')
        return res

    def insert_new_warn(self, latest_alarm, inter_part):
        new_warn = self.get_diff_part(latest_alarm, inter_part, latest_alarm)
        if not new_warn.empty:
            res = new_warn.rename(columns={'data_time': 'start_time'})
            res['dev_type'] = 3
            res['warn_type'] = 4
            res['rule_number'] = 2
            res['dev_type_name'] = '治污设施'
            res['release_status'] = 0
            res['end_time'] = self.data_time
            res['duration'] = res.apply(lambda x:self.cal_duration(x), axis=1)
            res['warn_level'] = res.apply(lambda x:self.cal_warn_level(x), axis=1)
            res['other_data'] = res.apply(lambda x:json.dumps({'yield_dev_id_list':x['yield_dev_id_list']}), axis=1)
            res.drop(columns='yield_dev_id_list', inplace=True)
            print(res)
            self._df_res_to_db(res)

    def update_unreleased_warn(self, unreleased_warn, inter_part, latest_alarm):
        closing_warn = self.get_diff_part(unreleased_warn, inter_part, unreleased_warn)
        if not closing_warn.empty:
            closing_warn['release_status'] = 1
            closing_warn['rule_number'] = 2
            closing_warn['warn_type'] = 4
            print(closing_warn)
            self._df_res_replace_to_db(closing_warn)

        unreleased_warn = pandas.merge(unreleased_warn, inter_part, on=['ent_id', 'county_id', 'dev_id'], how='right')
        if not unreleased_warn.empty:
            unreleased_warn['end_time'] = self.data_time
            unreleased_warn['rule_number'] = 2
            unreleased_warn['warn_type'] = 4
            unreleased_warn['duration'] = unreleased_warn.apply(lambda x:self.cal_duration(x), axis=1)
            unreleased_warn['warn_level'] = unreleased_warn.apply(lambda x:self.cal_warn_level(x), axis=1)

            try:
                for idx, v in inter_part.iterrows():
                    other_data = unreleased_warn[unreleased_warn['dev_id'] == v['dev_id']]['other_data'].tolist()[0]
                    new_yield_dev_id = latest_alarm[latest_alarm['dev_id'] == v['dev_id']]['yield_dev_id_list'].tolist()[0]
                    if other_data:
                        other_data = json.loads(other_data)
                        if other_data.get('yield_dev_id_list'):
                            if isinstance(other_data['yield_dev_id_list'], list):
                                other_data['yield_dev_id_list'] += new_yield_dev_id
                            else:
                                other_data['yield_dev_id_list'] = new_yield_dev_id
                        else:
                            other_data.update({'yield_dev_id_list': new_yield_dev_id})
                    else:
                        other_data = {'yield_dev_id_list': new_yield_dev_id}
                    index = unreleased_warn[unreleased_warn['dev_id']==v['dev_id']].index
                    for i in index:
                        unreleased_warn.loc[i, 'other_data'] = json.dumps(other_data)
                print(unreleased_warn)
            except Exception as e:
                print(e)
            self._df_res_replace_to_db(unreleased_warn)

    def cal_duration(self, line):
        try:
            duration = datetime.datetime.strptime(line.end_time, '%Y-%m-%d %H:%M:%S') - line.start_time
        except Exception as e:
            print(e)
            duration = datetime.datetime.strptime(line.end_time, '%Y-%m-%d %H:%M:%S') - datetime.datetime.strptime(line.start_time, '%Y-%m-%d %H:%M:%S')
        day = duration.days
        hour = duration.seconds // 3600
        minute = round(duration.seconds % 3600 / 60)
        dur_str = f'''{day}天{hour}小时{minute}分''' if day else f'''{hour}小时{minute}分'''
        return dur_str

    def cal_warn_level(self,line):
        try:
            duration = datetime.datetime.strptime(line.end_time, '%Y-%m-%d %H:%M:%S') - line.start_time
        except Exception as e:
            duration = datetime.datetime.strptime(line.end_time, '%Y-%m-%d %H:%M:%S') - datetime.datetime.strptime(line.start_time, '%Y-%m-%d %H:%M:%S')
        h = duration.days * 24 + duration.seconds / 3600
        if h > 4:
            return 1
        elif h > 2 and h <= 4:
            return 2
        elif h <= 2 and h >= 0.5:
            return 3
        else:
            return 0


    def _db_util(self):
        database_config = {
            'db_name': 'electricity_data',
            'username': 'electricity_api_service',
            'password': 'GJlfh7&#jg',
            'host': 'mmservice-05.mysql.hotgrid.cn',
            'port': 3306
        }
        conn = pymysql.connect(host=database_config['host'], user=database_config['username'],
                               password=database_config['password'], database=database_config['db_name'])
        cur = conn.cursor()
        return cur, conn

    def _db_res_to_df(self, sql):
        self.conn.ping()
        res_pd = pandas.read_sql(sql, con=self.conn)
        self.conn.close()
        return res_pd

    def _df_res_to_db(self, pd):
        if platform.system() == 'Linux':
            conn1 = 'mysql+pymysql://electricity_data:EF2zUl1GHss3yqay@192.168.195.201:3317/electricity_data'
        else:
            conn1 = 'mysql+pymysql://electricity_data:EF2zUl1GHss3yqay@10.20.7.227:33176/electricity_data'
        conn2 = 'mysql+pymysql://electricity_api_service:GJlfh7&#jg@mmservice-05.mysql.hotgrid.cn:3306/electricity_data'

        def mysql_replace_into(table, conn, keys, data_iter):
            from sqlalchemy.dialects.mysql import insert
            from sqlalchemy.ext.compiler import compiles
            from sqlalchemy.sql.expression import Insert

            @compiles(Insert)
            def replace_string(insert, compiler, **kw):
                s = compiler.visit_insert(insert, **kw)
                s = s.replace("INSERT INTO", "REPLACE INTO")
                return s

            data = [dict(zip(keys, row)) for row in data_iter]
            conn.execute(table.table.insert(replace_string=""), data)
        pd['warn_type'] = 4
        # pd = pd[['start_time', 'ent_id', 'warn_type', 'dev_id', 'dev_type', 'dev_type_name',
        #          'release_status',
        #          'dev_name', 'ent_name', 'industry_type_id', 'industry_type', 'county_id', 'county_name', 'town_id',
        #          'town_name',
        #          # 'end_time','duration',
        #          ]]
        redis_store = redis.Redis(host='electricity-service.redis.hotgrid.cn', port=6379, db=2,
                                  password='phWOgP05ymg01GaR')
        pd['code'] = pd['dev_name'].apply(lambda x: 'EB' + '{:0>14}'.format(redis_store.incr('elec_warn_code_incr', 1)))


        try:
            pd.to_sql(
                name='elec_warn_records',
                con=conn1,
                index=False,
                chunksize=500,
                if_exists='append', method=mysql_replace_into)
        except:
            pass
        try:
            pd.to_sql(
                name='elec_warn_records',
                con=conn2,
                index=False,
                chunksize=500,
                if_exists='append', method=mysql_replace_into)
        except:
            pass


        # pd.to_sql(name='elec_warn_records',
        #           con='mysql+pymysql://electricity:X1w4G3Hdl7VCNqhs@36.110.47.24:3317/electricity_data_test',
        #           index=False,
        #           chunksize=500,
        #           if_exists='append')
        print('*' * 30, 'kcs向测试数据库的elec_warn_records保存数据')

    def _df_res_replace_to_db(self, pd):
        if platform.system() == 'Linux':
            conn1 = 'mysql+pymysql://electricity_data:EF2zUl1GHss3yqay@192.168.195.201:3317/electricity_data'
        else:
            conn1 = 'mysql+pymysql://electricity_api_service:GJlfh7&#jg@mmservice-05.mysql.hotgrid.cn:3306/electricity_data'
        conn2 = 'mysql+pymysql://electricity_api_service:GJlfh7&#jg@mmservice-05.mysql.hotgrid.cn:3306/electricity_data'

        def mysql_replace_into(table, conn, keys, data_iter):
            from sqlalchemy.dialects.mysql import insert
            from sqlalchemy.ext.compiler import compiles
            from sqlalchemy.sql.expression import Insert

            @compiles(Insert)
            def replace_string(insert, compiler, **kw):
                s = compiler.visit_insert(insert, **kw)
                s = s.replace("INSERT INTO", "REPLACE INTO")
                return s

            data = [dict(zip(keys, row)) for row in data_iter]

            conn.execute(table.table.insert(replace_string=""), data)
        try:
            pd.to_sql(
                name='elec_warn_records',
                con=conn1,
                index=False,
                chunksize=500,
                if_exists='append', method=mysql_replace_into)
        except:
            pass
        try:
            pd.to_sql(
                name='elec_warn_records',
                con=conn2,
                index=False,
                chunksize=500,
                if_exists='append', method=mysql_replace_into)
        except:
            pass
        # 保存到测试数据库
        # pd.to_sql(name='elec_warn_records',
        #        con='mysql+pymysql://electricity:X1w4G3Hdl7VCNqhs@36.110.47.24:3317/electricity_data_test',
        #        index=False,
        #        chunksize=500,
        #        if_exists='append', method=mysql_replace_into)
        print('*' * 30, 'kcs向测试数据库的elec_warn_records保存数据')


if __name__ == '__main__':
    da = DailyAlarmNewRule2()
    # import platform
    # if platform.system() == 'Windows':
    #     for i in range(2):
    #         da.run(f"225,{9 - i},0")
    # else:
    #     # 01月9号到 今天的报警数据 从凌晨6点开始会算
    #     import pandas as pd
    #     # 定义起始和结束日期
    #     start_date = "2024-01-02"
    #     end_date = "2024-01-22"
    #     # 生成时间序列，从凌晨6点开始，每隔15分钟
    #     date_time_list = pd.date_range(start=start_date, end=end_date, freq='15min')
    #     # 仅选择从凌晨6点开始的时间段
    #     date_time_list = [dt for dt in date_time_list if dt.hour >= 11]
    #     # 将datetime对象转化为字符串并存储在新的列表中
    #     date_time_str_list = [dt.strftime('%Y-%m-%d %H:%M:%S') for dt in date_time_list]
    #     for i in range(len(date_time_str_list)):
    #         da.run(f"225,{9 - i},{date_time_str_list[i]}")
    import pandas as pd
    # 删除表的数据信息
    # 定义起始和结束日期
    start_date = "2024-01-02"
    end_date = "2024-01-22"
    # 生成时间序列，从凌晨6点开始，每隔15分钟
    date_time_list = pd.date_range(start=start_date, end=end_date, freq='15min')
    # 仅选择从凌晨6点开始的时间段
    date_time_list = [dt for dt in date_time_list if dt.hour >= 11]
    # 将datetime对象转化为字符串并存储在新的列表中
    date_time_str_list = [dt.strftime('%Y-%m-%d %H:%M:%S') for dt in date_time_list]
    for i in range(len(date_time_str_list)):
        # da.delete_data_elec_daily_alarm_rule2(date_time_str_list[i])
        da.delete_data_elec_warn_records(date_time_str_list[i])